import MySQLdb
import ConfigParser
import string, os, sys
import time
import logging 
#from pykafka import KafkaClient
from kafka.producer import SimpleProducer  
from kafka.client import KafkaClient 
from logging.handlers import TimedRotatingFileHandler

loggerName = 'indexDataFinal'
logFileName = loggerName + '.log'
logFileDirectory = '/data2/logs/fetchIndexData'
logFilePath = logFileDirectory + '/' + logFileName
logWhen = 'D'
logInterval = 1

formatter = logging.Formatter('%(asctime)s %(levelname)s %(name)s %(funcName)s %(threadName)s %(message)s')

handler = TimedRotatingFileHandler(logFilePath, when=logWhen, interval=logInterval, backupCount=5)
handler.setFormatter(formatter)

logger = logging.getLogger(loggerName)
logger.setLevel(logging.INFO)
logger.addHandler(handler)


class KafkaMessage:
  def __init__(self,indexS,typeS,entityIdsS,stageS,time):
    self.index = indexS
    self.type = typeS
    self.entityIds = entityIdsS
    self.stage = stageS
    self.time = time
  def getIndex(self):
    return self.index
  def getType(self):
    return self.type
  def getEntityIds(self):
    return self.entityIds
  def getStage(self):
    return self.stage
  def getTime(self):
    return self.time
  def setIndex(self, indexS):
    self.index = indexS
  def setType(self, typeS):
    self.type = typeS
  def setEntityIds(self, entityIdsS):
    self.entityIds = entityIdsS
  def setStage(self, stageS):
    self.stage = stageS
  def setStage(self, time):
    self.time = time

  def toJson(self):
    d = {'_index':self.index,'_type':self.type,'_entityIds':self.entityIds,'_stage':self.stage,'_time':self.time}
    return str(d)

def mysql(cur, sql):
  try:
    time_need = time.time()
    count=cur.execute(sql)
    print 'there has %s rows record' % count
    logger.info('there has %s rows record' % count)
    ret = [] 
    results=cur.fetchall()
    i = 0
    if results:
      for r in results:
        print r
        logger.info(r)
        ret.append(int(r[0]))
        i = i + 1
        print r[0]
        logger.info(r[0])
    return ret,int(time_need)
  except MySQLdb.Error,e:
    print "Mysql Error %d: %s" % (e.args[0], e.args[1])
    logger.exception("Mysql Error %d: %s" % (e.args[0], e.args[1]))

def getFromTime(store_file):
  print store_file
  fileHandle = open(store_file, 'r')
  liness = ''
  try:
    for line in fileHandle:
      print line
      liness = line.strip()
    return liness
  except Exception, e:
    raise e
  finally:
    fileHandle.close()

def producer_send(topic, km, producer):
  try:
    producer.send_messages(topic, km.toJson())
    return 1
  except Exception,sss:
    print sss
    logger.exception(str(sss))
    return 2

def send_messages(index, typeS, stage, batch_send, producer1, entities, topic):
    
    producer = producer1
    i = 0
    j = 0
    jj = len(entities)
    entities_send = []
    for entity in entities:
      j = j + 1
      entities_send.append(entity)
      i = i + 1
      
      if i >= batch_send or j == jj:
        t_now = int(time.time()*1000)
        km = KafkaMessage(index, typeS, entities_send, stage, t_now)
        p_result = 2
        count = 0
        sendStr = str(entities_send)
        while (p_result == 2):
          if count > 0:
            logger.info("error:%s,count:%s" % (sendStr,str(count)))
            time.sleep(1*count)
          p_result = producer_send(topic, km, producer)
          if p_result == 2:
            count = count + 1
        logger.info("message send successfully:: index:%s, type:%s, entities_send:%s, stage:%d." % (index,typeS,sendStr,stage))
        entities_send = []
        i = 0
        count = 0
        #try:
          #producer.send_messages(topic, km.toJson())
          #logger.info('producer is ok')
          #entities_send = []
          #i = 0
        #except Exception,sss:
          #print sss
          #logger.error(str(sss))

    stage = stage + 1
    return stage

def getKafkaProducer(hosts):
  client=KafkaClient(hosts)
  return SimpleProducer(client,batch_send=False,batch_send_every_n=1)
def main(argv):
  agv_len = len(argv)
  if agv_len == 1:
    file_name = '/data2/logs/fetchIndexData/configNew1'
    store_file = '/data2/logs/fetchIndexData/fromTimeNew1'
  else:
    file_name = argv[1]
    store_file = argv[2]
  
  cf = ConfigParser.ConfigParser()  
  cf.read(file_name)
  db_info_host = cf.get("db.info", "db_info_host")  
  db_info_port = cf.getint("db.info", "db_info_port")  
  db_info_user = cf.get("db.info", "db_info_user")  
  db_info_pass = cf.get("db.info", "db_info_pass")
  db_tag_host = cf.get("db.tag", "db_tag_host")  
  db_tag_port = cf.getint("db.tag", "db_tag_port")  
  db_tag_user = cf.get("db.tag", "db_tag_user")  
  db_tag_pass = cf.get("db.tag", "db_tag_pass")
  kafka_hosts = cf.get("kafka", "hosts")
  topic = cf.get("kafka", "topic")
  print 'TOPIC=',topic

  cyc_time = cf.getint("common", "cyc_time")
  index = cf.get("common", "index")
  typeS = cf.get("common", "type")
  batch_num_init = cf.getint("common", "batch_num_init")
  batch_num_common = cf.getint("common", "batch_num_common")
 
  timeandstage = getFromTime(store_file)
  print timeandstage
  logger.info(timeandstage)
  lins = timeandstage.split(':')
  fromTime = int(lins[0])
  stage = int(lins[1])

  sql1 = "select distinct id from qyer_lastminute.lastminute_main where title != '' and status = 0;"
  sql2 = "select distinct lid from tag.lastminute_tag_mapping where tag_name != '' and last_modified_time >= "
  sql3 = "select distinct lid from tag.lastminute_tag_mapping where tag_name != '' and last_modified_time >= "
  sql4 = "select count(1) from tag.lastminute_tag_mapping;"
  sql5 = "select distinct id from qyer_lastminute.lastminute_main where title != '' and status = 0 and modifytime >= "
  sql6 = "select distinct id from qyer_lastminute.lastminute_main where title != '' and status = 0 and modifytime >= "
  fileHandle = open(store_file, 'w')
  
  try:
    producer = getKafkaProducer(kafka_hosts)
    conn_tag = MySQLdb.connect(host=db_tag_host, user=db_tag_user, passwd=db_tag_pass,port=db_tag_port)
    conn_info = MySQLdb.connect(host=db_info_host, user=db_info_user, passwd=db_info_pass,port=db_info_port)
    cur_tag = conn_tag.cursor()
    cur_info = conn_info.cursor()
  except Exception,ex:
    print ex
    logger.exception(str(ex))
    sys.exit(-11)

  try:
    fileHandle.close()
    while True:
      try:
        logger.info("begin stage: %s" % stage)
        conn_tag.ping()
        conn_info.ping()
      except Exception,esss:
        print esss
        logger.exception(str(esss))
        try:
          conn_tag = MySQLdb.connect(host=db_tag_host, user=db_tag_user, passwd=db_tag_pass,port=db_tag_port)
          conn_info = MySQLdb.connect(host=db_info_host, user=db_info_user, passwd=db_info_pass,port=db_info_port)
          cur_tag = conn_tag.cursor()
          cur_info = conn_info.cursor()
        except Exception,essss:
          print essss
          logger.exception(str(essss))
          time.sleep(cyc_time*2)
          continue
      try:
        if fromTime == 0:
          entities,fromTime = mysql(cur_info, sql1)
        else:
          sql5 = sql6
          sql5 = sql5 + str(fromTime) + ';'
          print sql5
          logger.info("info sql:"+sql5)
          entities2,fromTime3 = mysql(cur_info, sql5)
          conn_info.commit()
          logger.info("tag Lid list:%s" % str(entities2))

          sql2 = sql3
          sql2 = sql2 + str(fromTime) + ';'
          print sql2
          logger.info("tag sql:"+sql2)
          entities1,fromTime1 = mysql(cur_tag, sql2)
          conn_tag.commit()
          logger.info("info Lid list:%s" % str(entities1))
          
          entities = list(set(entities2).union(set(entities1)))
          logger.info("Lid list:%s" % str(entities))
          fromTime = int(fromTime3) - 5
        if entities:
          if stage == 0:
            stage = send_messages(index, typeS, stage, batch_num_init, producer, entities, topic)
          else:
            stage = send_messages(index, typeS, stage, batch_num_common, producer, entities, topic)
        fileHandle = open(store_file, 'w')
        fileHandle.write(str(fromTime) + ":" + str(stage))
        fileHandle.close()
        time.sleep(cyc_time)
      except Exception,ess:
        print ess
        logger.exception(str(ess))
  except Exception,es:
    print es
    logger.exception(str(es))
    #sys.exit(-11)
  finally:
    producer.stop(1)
    fileHandle.close()
    cur_tag.close()
    cur_info.close()
    conn_tag.close()
    conn_info.close()

if __name__ == '__main__':
  main(sys.argv)

